package ct.analysis.io;

import ct.common.util.JDBCUtil;
import ct.analysis.kv.AnalysisKey;
import ct.analysis.kv.AnalysisValue;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author Jeremy Zheng
 * @Date 2021/2/26 20:55
 * @Version 1.0
 * MySQL数据格式化输出对象
 */
public class MySQLBeanOutputFormat extends OutputFormat<AnalysisKey, AnalysisValue> {

    protected static class MySQLRecordWriter extends RecordWriter<AnalysisKey, AnalysisValue> {

        private Connection connection=null;
        Map<String,Integer> userMap=new HashMap<>();
        Map<String,Integer> dateMap=new HashMap<>();

        public MySQLRecordWriter(){
            //获取资源
            connection= JDBCUtil.getConnection();
            PreparedStatement pstat = null;
            ResultSet rs =null;
            try {
                //读取用户，时间的数据
                String queryUserSql="select id,tel from ct_user";
                pstat=connection.prepareStatement(queryUserSql);
                rs=pstat.executeQuery();
                while (rs.next()){
                    Integer id = rs.getInt(1);
                    String tel=rs.getString(2);
                    userMap.put(tel,id);
                }
                rs.close();

                String queryDateSql="select id,year,month,day from ct_date";
                pstat=connection.prepareStatement(queryDateSql);
                rs=pstat.executeQuery();
                while (rs.next()){
                    Integer id = rs.getInt(1);
                    String year=rs.getString(2);
                    String month=rs.getString(3);
                    String day=rs.getString(4);
                    if (month.length()==1){
                        month="0"+month;
                    }
                    if (day.length()==1){
                        day="0"+day;
                    }
                    dateMap.put(year+month+day,id);
                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                if(pstat!=null){
                    try {
                        pstat.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
                if (rs!=null){
                    try {
                        rs.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }



        }

        //输出数据
        @Override
        public void write(AnalysisKey key, AnalysisValue value) throws IOException, InterruptedException {

            PreparedStatement pstat=null;
            try {
                String insertSQL="insert into ct_call(telid,dateid,sumcall,sumduration) values(?,?,?,?)";
                pstat=connection.prepareStatement(insertSQL);

                pstat.setInt(1,userMap.get(key.getTel()));
                pstat.setInt(2,dateMap.get(key.getDate()));
                pstat.setInt(3,Integer.parseInt(value.getSumCall()));
                pstat.setInt(4,Integer.parseInt(value.getSumDuration()));
                pstat.executeUpdate();
            } catch (SQLException e) {
                e.printStackTrace();
            }finally {

                if (pstat!=null){
                    try {
                        pstat.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }


            }
        }

        //释放资源
        @Override
        public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            if (connection!=null){
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

        }

    }

    @Override
    public RecordWriter<AnalysisKey, AnalysisValue> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new MySQLRecordWriter();
    }

    @Override
    public void checkOutputSpecs(JobContext jobContext) throws IOException, InterruptedException {

    }


    private FileOutputCommitter committer = null;

    public static Path getOutputPath(JobContext job) {
        String name = job.getConfiguration().get("mapreduce.output.fileoutputformat.outputdir");
        return name == null ? null : new Path(name);
    }

    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        if (this.committer == null) {
            Path output = getOutputPath(taskAttemptContext);
            this.committer = new FileOutputCommitter(output, taskAttemptContext);
        }

        return this.committer;
    }
}

